/* 
 * Copyright [2018] [Alex/libo(liboware@gmail.com)]
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *  http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.hyts.reactor.produce;

import com.hyts.reactor.base.ProduceManager;
import com.hyts.reactor.constant.ReactorType;
import com.hyts.reactor.handler.ReactorEventHandler;
import com.hyts.reactor.produce.DisruptorProduceManager.User;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;

import lombok.Getter;
import reactor.core.Environment;
import reactor.core.Reactor;
import reactor.core.spec.Reactors;
import reactor.event.Event;
import reactor.event.registry.Registration;
import reactor.event.selector.Selectors;
import reactor.function.Consumer;

/** 
 * @author LiBo/Alex
 * @version V1.0 
 */
public class SpringReactorProduceManager<T> extends ProduceManager<T> {
	
	/**  
	 * @fieldName env
	 * @fieldType Environment
	 */ 
	@Getter
	private Environment env;
	
	/**  
	 * @fieldName reactorType
	 * @fieldType ReactorType
	 */ 
	@Getter
	private ReactorType reactorType;
	
	/**  
	 * @fieldName topic
	 * @fieldType String
	 */ 
	@Getter
	private String topic;
	
	/**  
	 * @fieldName reactor
	 * @fieldType Reactor
	 */ 
	@Getter
	private Reactor reactor;

	/**  
	 * @constructor��ReactorProduceManager  
	 * @param env
	 * @param reactorType
	 * @param topic  
	 */ 
	public SpringReactorProduceManager(Environment env, ReactorType reactorType, String topic) {
		super();
		this.env = env;
		this.reactorType = reactorType;
		this.topic = topic;
		reactor = Reactors.reactor().env(env).dispatcher(reactorType.toString()).get();
	}

	/**  
	 * @constructor��ReactorProduceManager  
	 * @param env
	 * @param topic  
	 */ 
	public SpringReactorProduceManager(Environment env, String topic) {
		super();
		this.env = env;
		this.reactorType = ReactorType.THREAD_POOL;
		this.topic = topic;
		reactor = Reactors.reactor().env(env).dispatcher(reactorType.toString()).get();
	}

	/**  
	 * @constructor��ReactorProduceManager  
	 * @param reactorType
	 * @param topic  
	 */ 
	public SpringReactorProduceManager(ReactorType reactorType, String topic) {
		super();
		this.env = new Environment();
		this.reactorType = reactorType;
		this.topic = topic;
		reactor = Reactors.reactor().env(env).dispatcher(reactorType.toString()).get();
	}
	
	/**  
	 * @param topic
	 */ 
	public SpringReactorProduceManager(String topic) {
		super();
		this.env = new Environment();
		this.reactorType = ReactorType.THREAD_POOL;
		this.topic = topic;
		reactor = Reactors.reactor().env(env).dispatcher(reactorType.toString()).get();
	}
	
	/**  
	 * @return
	 * @exception
	 */ 
	public Reactor product(T param) {
		return reactor.notify(topic,Event.wrap(param));
	}
	
	/**  
	 * @return
	 * @exception
	 */ 
	public Reactor addHandler(ReactorEventHandler<T> handler){
		reactor.on(Selectors.$(topic),handler::handleEvent);
		return reactor;
	}
	
	
	public static class User implements EventFactory{
		
		public String name;
		
		/* (�� Javadoc)
		 * @return
		 * @see com.lmax.disruptor.EventFactory#newInstance()  
		 * @exception
		 */ 
		
		
		@Override
		public User newInstance() {
			return new User();
		}

		public String getName() {
			return name;
		}

		public void setName(String name) {
			this.name = name;
		}
	}
	
	
	public static void main(String[] args) {
		SpringReactorProduceManager<User> manager = new SpringReactorProduceManager<User>("test");
		/*manager.build(new User() {{
			this.setName("1231");
		}});*/
		manager.addHandler((event)->{
			System.out.println(event.getData().getName());
		});
		for (long l = 0; true; l++) {
			String obj = "Test-" + l;
			manager.product(new User() {{
				this.setName(obj);
			}});
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}
	
}
